﻿//  EventSubscription 负责处理本端向远端订阅的消息.

using System;
using System.Threading;
using Google.Protobuf;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using GSE.Rpc;

namespace GSE
{
    //  public MsgType 必须实现Google.Protobuf.IMessage
    public interface IEventConsumer<MsgType>
    {
        void Process(MsgType[] messages);
    }

    /// <summary>
    /// 当取消对远端消息的订阅时.
    /// </summary>
    /// <param name="sub"></param>
    public delegate void OnCancelSub<MessageType>(EventSubscription<MessageType> sub);

    public class EventIncomingDispatcher
    {
        /// <summary>
        /// 订阅信息表.
        /// </summary>
        protected ConcurrentDictionary<UInt16, Object> eventSubscriptions;

        protected BlockingCollection<Push> inComingEventQueue;
        protected ILogger logger
        {
            get { return Logger.Default; }
        }
        public EventIncomingDispatcher()
        {
            eventSubscriptions = new ConcurrentDictionary<ushort, Object>();
            inComingEventQueue = new BlockingCollection<Push>(3000);
            Task.Run(() => dispatch());
        }

        public void RegisterSub<MessageType>(EventSubscription<MessageType> sub)
        {
            eventSubscriptions[sub.SeriesID] = sub;

            logger.Info($"Registered event subscription:{sub}");
        }

        public void UnRegisterSub<MessageType>(EventSubscription<MessageType> sub)
        {
            eventSubscriptions.TryRemove(sub.SeriesID, out _);
        }

        public void ReceivePush(Push p)
        {
            logger.Debug($"received event push， seriesID {p.seriesID}");
            inComingEventQueue.Add(p);
        }

        protected void dispatch()
        {
            while (true)
            {
                var p = inComingEventQueue.Take();
                var exists = eventSubscriptions.TryGetValue(p.seriesID, out var s);
                if (!exists)
                {
                    logger.Warn($"No subscriptions for incoming message. seriesID: {p.seriesID}");
                }
                else
                {
                    try
                    {
                        s.GetType().GetMethod("Consume").Invoke(s, new object[] { p });
                    }
                    catch (Exception ex)
                    {
                        logger.Error($"event dispatch exception: {ex.Message} {ex.StackTrace}");
                        Console.WriteLine($"EEEEE{ex.ToString()}, {ex.StackTrace} [{s == null}]");
                    }
                }
            }
        }
    }

    /// <summary>
    /// 本端向远端发起的消息/事件订阅处理器.
    /// </summary>
    public class EventSubscription<MessageType>
    {
        protected byte seriesID;
        protected IEventConsumer<MessageType> consumer;
        protected OnCancelSub<MessageType> OnCancel;
        protected BlockingCollection<Push> eventDataQueue;
        protected CancellationTokenSource cts;
        protected int canceled = 0;

        public CancellationToken ctsToken
        {
            get { return cts.Token; }
        }

        protected ILogger logger
        {
            get { return Logger.Default; }
        }

        public byte SeriesID
        {
            get { return seriesID; }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="subRsp"></param>
        /// <param name="consumer"></param>
        /// <param name="onCancel">调用Cancel方法之前会调用onCancel</param>
        public EventSubscription(protos.SubEventResponse subRsp, IEventConsumer<MessageType> consumer, OnCancelSub<MessageType> onCancel)
        {
            seriesID = (byte)subRsp.SeriesID;
            cts = new CancellationTokenSource();
            this.consumer = consumer;
            this.OnCancel = onCancel;
            eventDataQueue = new BlockingCollection<Push>(1000);
            Task.Run(() =>
            {
                while (true)
                {
                    try
                    {
                        var push = eventDataQueue.Take(ctsToken);
                        consumer.Process(push.DecodeMessages<MessageType>());
                    }
                    catch (OperationCanceledException)
                    {
                        logger.Debug("event consuming is cancled.");
                        return;
                    }
                    catch (Exception ex)
                    {
                        logger.Warn($"event consuming exception: {ex.Message}, canceling... ");
                        Cancel();
                        return;

                    }
                }
            }, ctsToken);
        }

        public void Cancel()
        {
            var i = Interlocked.Exchange(ref canceled, 1);
            if (i == 1)
            {
                return;
            }
            OnCancel(this);
            cts.Cancel();
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="rawEventData">protobuf 的字节码</param>
        public void Consume(Push push)
        {
            eventDataQueue.Add(push, ctsToken);

        }

    }
}
